#include "BigQ.h"

void BigQ::Wait(){
    pthread_join(thread, NULL);
}

BigQ::BigQ (Pipe &in, Pipe &out, OrderMaker &sortorder, int runlen):in(in), out(out), sortorder(sortorder), runlen(runlen){
    records.clear();
    noOfRuns = 0;
    currentPage = 0;
    noOfPages = 0;

    //produce unique file name for run file
    char *dir = NULL;
    FILE *fp = fopen ("test.cat", "r");
	if (fp) {
		char *mem = (char *) malloc (80);
		dir = &mem[0];
		char line[80];
		fgets (line, 80, fp);
		fgets (line, 80, fp);
		sscanf (line, "%s\n", dir);
		fclose (fp);
	}
	else
        dir = "";

    int randNumber = rand();
    sprintf(tempPath, "%srun%d.temp", dir, randNumber);

    pthread_create(&thread, NULL, &Worker, (void *) this);
}

BigQ::~BigQ () { pthread_join(thread, NULL); }

void BigQ::RunToFile(){
    noOfPages = 0;
    page.EmptyItOut();
    bool clear = true;
    if (records.size() > 0){
        sort (records.begin(), records.end(), comparator);
        for(vector<Record*>::iterator iter = records.begin(); iter!= records.end(); iter++){
            if (!page.Append(*iter)){
                file.AddPage(&page, currentPage);
                currentPage++;
                noOfPages++;
                page.EmptyItOut();
                page.Append(*iter);
            }
        }

        if (noOfPages == runlen){
            clear = false;
            records.clear();
            Record * r = new Record;
            Record * recordCopy;

            while (page.GetFirst(r)){
                recordCopy = new Record;
                recordCopy->Copy(r);
                records.push_back(recordCopy);
            }

            for(vector<Record*>::iterator iter = records.begin(); iter!= records.end(); iter++){
                recordCopy = new Record;
                recordCopy->Copy(*iter);
                page.Append(recordCopy);
            }
        }
        else{
            file.AddPage(&page, currentPage);
            currentPage++;
            noOfPages++;
            page.EmptyItOut();
        }

        noOfRuns++;
    }
    if (clear)
        records.clear();
}

void BigQ::Sort () {
    Record * record = new Record;
    Record * recordCopy;
    comparator.order = sortorder;

    int nOfPages = 0;
    file.Open(0, tempPath);

    // read data from in pipe sort them into runlen pages
    while (in.Remove(record)){
        recordCopy = new Record;
        recordCopy->Copy(record);

        if (page.Append(record)){
            records.push_back(recordCopy);
        }
        else{
            nOfPages++;

            if (nOfPages == runlen){
                RunToFile();
                nOfPages = 0;
            }
            else
                page.EmptyItOut();

            page.Append(record);
            records.push_back(recordCopy);
        }
    }

    RunToFile();

    if (records.size() > 0)
        RunToFile();

    records.clear();
    page.EmptyItOut();

    // construct priority queue over sorted runs and dump sorted data
    // into the out pipe
    pq_type pq (comparator);

    int runLimits[noOfRuns];
    int nextRunPage[noOfRuns];
    Page pages[noOfRuns];

    for (int i = 0; i < noOfRuns; i++){
        file.GetPage(&pages[i], (i * runlen));
        nextRunPage[i] = (i * runlen) + 1;
        runLimits[i] = 1;
        Record * r = new Record;
        pages[i].GetFirst(r);
        RunRecord * rr = new RunRecord(r, i);
        pq.push(rr);
    }

    while (!pq.empty()){
        RunRecord * rr = pq.top();
        int currentRun = rr->run;
        pq.pop();
        out.Insert(rr->record);

        Record * r = new Record;
        if (pages[currentRun].GetFirst(r)){
            RunRecord * rr = new RunRecord(r, currentRun);
            pq.push(rr);
        }
        else{
            int limit;
            if (currentRun == noOfRuns - 1)
                limit = (file.GetLength() - 1) - (currentRun * runlen);
            else
                limit = runlen;

            if (runLimits[currentRun] != limit){
                file.GetPage(&pages[currentRun], nextRunPage[currentRun]);
                nextRunPage[currentRun]++;
                runLimits[currentRun]++;
                Record * r = new Record;
                pages[currentRun].GetFirst(r);
                RunRecord * rr = new RunRecord(r, currentRun);
                pq.push(rr);
            }
        }
    }

    // finally shut down the out pipe
	out.ShutDown ();
	file.Close();
	delete record;
	delete recordCopy;
	remove(tempPath);
}

void *Worker(void *param){
    BigQ *bq = (BigQ *) param;
    bq->Sort();
	pthread_exit(NULL);
}

RunRecord::RunRecord(Record* record, int run):record(record), run(run){}
